package com.vf.cloud.paas.app.master.common.server.ueManager.hander;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.jfinal.kit.JsonKit;
import com.jfinal.kit.StrKit;
import com.vf.cloud.paas.app.master.api.scene.service.*;
import com.vf.cloud.paas.app.master.common.domain.SceneAction;
import com.vf.cloud.paas.common.cirrus.Cirrus;
import com.vf.cloud.paas.common.service.IRedis;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpMessage;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpUtil;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.CharsetUtil;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class UEManagerHandler extends SimpleChannelInboundHandler<Object> {

	private IRedis redisImpl;
	private ISceneAction sceneActionImpl;

	public UEManagerHandler(IRedis redisImpl, ISceneAction sceneActionImpl) {
		super();
		this.redisImpl = redisImpl;
		this.sceneActionImpl = sceneActionImpl;
	}

	// 用于记录和管理所有客户端的channel
	public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		log.info(String.format("UE connected:" + ctx.channel().remoteAddress()));
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		ctx.channel().writeAndFlush(new TextWebSocketFrame("disconnected"));
		log.error(String.format("UE disconnected:" + ctx.channel().remoteAddress()));
		
		String token=redisImpl.get("Token_"+ctx.channel().id().asLongText());
		if(!StrKit.isBlank(token)) {
			redisImpl.delete("Token_"+ctx.channel().id().asLongText());
			Cirrus.UE_Client.remove(token);
		}
		
		Cirrus.UE_Client.remove(redisImpl.get(ctx.channel().id().asLongText()));
		redisImpl.delete(ctx.channel().id().asLongText());
		redisImpl.delete(redisImpl.getByForRunCmd(ctx.channel().id().asLongText()));
		ctx.close();
	}

	@Override
	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
		ctx.flush();
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		ctx.close();
		//cause.printStackTrace();
	}

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
		if (msg instanceof FullHttpRequest) {
			handleHttpRequest(ctx, (FullHttpRequest) msg);
		} else {
			handlerWebSocketFrame(ctx, msg);
		}
	}

	protected WebSocketServerHandshaker handshaker;

	/**
	 * 处理 FullHttpRequest
	 * 
	 * @param ctx
	 * @param request
	 */
	protected void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {
		if (!req.decoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) {
			sendHttpResponse(ctx, req,
					new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
			return;
		}
		String uri = req.uri();
		WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
				"ws://" + req.headers().get(HttpHeaderNames.HOST) + uri, null, false);
		handshaker = wsFactory.newHandshaker(req);
		if (handshaker == null) {
			WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
		} else {
			handshaker.handshake(ctx.channel(), req);
		}

		uri = uri.replaceFirst("\\/", "");
		String[] params = uri.split("\\/", -1);
		if (params == null || params.length <= 0) {
			ctx.close();
			return;
		}

		String appId = params[0];
		if (StrKit.isBlank(appId)) {
			ctx.close();
			return;
		}
		//Cirrus.UE_Client.put(token, ctx);
		redisImpl.add(ctx.channel().id().asLongText(), appId);
	}

	public static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res) {
		if (res.status().code() != 200) {
			ByteBuf buf = Unpooled.copiedBuffer(res.status().code() + "", CharsetUtil.UTF_8);
			res.content().writeBytes(buf);
			buf.release();
		}
		// 如果是非Keep-Alive，关闭连接
		if (!HttpUtil.isKeepAlive((HttpMessage) req.protocolVersion()) || res.status().code() != 200) {
			ctx.channel().writeAndFlush(res).addListener(ChannelFutureListener.CLOSE);
		}
	}

	private void handlerWebSocketFrame(ChannelHandlerContext ctx, Object frame) {
		// 判断是否关闭链路的指令
		if (frame instanceof CloseWebSocketFrame) {
			CloseWebSocketFrame closeWebSocketFrame = (CloseWebSocketFrame) frame;
			handshaker.close(ctx.channel(), (CloseWebSocketFrame) closeWebSocketFrame.retain());
			return;
		}
		// 判断是否ping消息
		if (frame instanceof PingWebSocketFrame) {
			return;
		}

		if (frame instanceof TextWebSocketFrame) {
			TextWebSocketFrame textWebSocketFrame = (TextWebSocketFrame) frame;
			if (textWebSocketFrame.text().startsWith("40")) {
				String msg = textWebSocketFrame.text().substring(2);
				if (!StrKit.isBlank(msg)) {
					ctx.channel().writeAndFlush(new TextWebSocketFrame(textWebSocketFrame.text()));
					
					JSONObject object = JSONObject.parseObject(msg);
					if (object != null && object.containsKey("token")) {
						String token = object.getString("token");
						Cirrus.UE_Client.put(token, ctx);
						redisImpl.add("Token_"+ctx.channel().id().asLongText(), token);
						String appId=redisImpl.get(ctx.channel().id().asLongText());
						if(!StrKit.isBlank(appId)) {
							log.info(String.format("<- UE Token:%s -- AppId:%s", token,appId));
							SceneAction sceneAction = sceneActionImpl.getDefault(appId);
							if(sceneAction!=null) {
								run(ctx, sceneAction.getContent());
							}
						}
						this.APIAlready(ctx);
						
					} else {
						ctx.close();
						log.info(String.format("<- UE:Token is empty, so it is closed."));
					}
				}
				return;
			} else if (textWebSocketFrame.text().startsWith("42")) {
				String msg = textWebSocketFrame.text();
				msg = msg.substring(2);
				JSONArray array = JSONArray.parseArray(msg);
				if (array != null && array.size() == 2) {
					String method = array.getString(0);
					String dataJson = array.getString(1);
					handleFor42(ctx, method, dataJson);
				} else {
					log.error("unsupported UE message type:%s", msg);
				}
			} else {
				log.error("unsupported UE message type:%s", textWebSocketFrame.text());
			}
		}
	}

	/**
	 * 向客户端发送心跳
	 * 
	 * @param ctx
	 * @param evt
	 * @throws Exception
	 */
	@Override
	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
		if (evt instanceof IdleStateEvent) {
			IdleStateEvent event = (IdleStateEvent) evt;
			if (event.state().equals(IdleState.READER_IDLE)) {
				sendPing(ctx, System.currentTimeMillis());
			} else if (event.state().equals(IdleState.WRITER_IDLE)) {
				sendPing(ctx, System.currentTimeMillis());
			} else if (event.state().equals(IdleState.ALL_IDLE)) {
				sendPing(ctx, System.currentTimeMillis());
			}
		}
	}

	public void sendPing(ChannelHandlerContext ctx, long currentTimeMillis) {
		Map<String, Object> clientConfig = new HashMap<String, Object>();
		clientConfig.put("time", currentTimeMillis);
		// sendUE(ctx,"OnPaaS",clientConfig);
	}

	public void sendConnected(ChannelHandlerContext ctx, long currentTimeMillis) {
		Map<String, Object> clientConfig = new HashMap<String, Object>();
		clientConfig.put("type", "pong");
		clientConfig.put("time", currentTimeMillis);
		send(ctx, JsonKit.toJson(clientConfig));
	}

	public void sendUE(ChannelHandlerContext ctx, String method, Object data) {
		List<Object> list = new LinkedList<Object>();
		list.add(method);
		list.add(data);
		send(ctx, "42" + JsonKit.toJson(list));
	}

	public void sendOnPaaS(ChannelHandlerContext ctx, Object data) {
		List<Object> list = new LinkedList<Object>();
		list.add("OnPaaS");
		list.add(data);
		send(ctx, "42" + JsonKit.toJson(list));
	}

	public void APIAlready(ChannelHandlerContext ctx) {
		Map<String, Object> jsondata = new HashMap<String, Object>();
		jsondata.put("command", "APIAlready");
		jsondata.put("data", new HashMap<String, Object>());
		sendOnPaaS(ctx, jsondata);
	}

	public void run(ChannelHandlerContext ctx, String json) {
		if (!StrKit.isBlank(json)) {
			if (json.startsWith("{")) {
				JSONObject content = JSONObject.parseObject(json);
				if (content != null) {
					JSONObject data = content.getJSONObject("data");
					if (content.containsKey("event") && data != null) {
						JSONObject event = content.getJSONObject("event");
						if (event != null) {
							for (String key : event.keySet()) {
								System.out.println(key + ":" + event.get(key));
								JSONArray array = event.getJSONArray(key);
								if (array != null && array.size() > 0) {
									redisImpl.add(key + "_" + ctx.channel().id().asLongText(), array);
								}
							}
						}
					}
					if (data != null)
						this.sendOnPaaS(ctx, data);
				}
			} else {
				JSONArray array = JSONArray.parseArray(json);
				if (array != null && array.size() > 0) {
					for (int i = 0; i < array.size(); i++) {
						run(ctx, array.getString(i));
						System.out.println("----------------------------------------------");
					}
				}
			}
		}
	}

	public void handleFor42(ChannelHandlerContext ctx, String method, String dataJson) {
		String key = method + "_" + ctx.channel().id().asLongText();
		if (redisImpl.hasKey(key)) {
			String content = redisImpl.get(method + "_" + ctx.channel().id().asLongText());
			run(ctx, content);
			redisImpl.delete(method + "_" + ctx.channel().id().asLongText());
			log.info(String.format("<- UE content:%s", content));
		}
	}

	public void send(ChannelHandlerContext ctx, String json) {
		if (ctx != null) {
			ctx.channel().writeAndFlush(new TextWebSocketFrame(json)).addListener(new ChannelFutureListener() {
				@Override
				public void operationComplete(ChannelFuture future) throws Exception {
					if (future.isSuccess()) {
						// log.info(String.format("-> Streamer:%s", json));
					} else {
						log.info(String.format("-> Streamer isError :%s", json));
					}
				}
			});
		}
	}

}
